#pragma once

#include <grpcpp/grpcpp.h>
#include <atomic>
#include <thread>
#include <vector>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <chrono>
#include <set>
#include <map>
#include "ingest.grpc.pb.h"
#include "sql_writer_pq.h"

using grpc::ServerContext;
using grpc::ServerReader;
using grpc::ServerReaderWriter;
using grpc::ServerWriter;
using grpc::Status;

using namespace kwdb::ingest::v1;

class StreamHandler : public IngestService::Service {
private:
    // Batch processing thresholds (configurable via environment)
    size_t batch_size_;
    int flush_interval_ms_;

    // Background writer thread (for future use)
    std::atomic<bool> running_{true};
    std::unique_ptr<std::thread> writer_thread_;

    // Thread-safe queue for batching
    std::queue<Point> point_queue_;
    std::mutex queue_mutex_;
    std::condition_variable queue_cv_;

    // Performance monitoring
    struct MonitoringStats {
        std::atomic<size_t> total_points{0};
        std::atomic<size_t> total_inserts{0};
        std::atomic<size_t> total_batches{0};
        std::atomic<size_t> queue_depth{0};
        std::chrono::steady_clock::time_point start_time;
        std::chrono::steady_clock::time_point last_report_time;
        std::atomic<size_t> points_since_last_report{0};
        std::atomic<size_t> inserts_since_last_report{0};

        MonitoringStats() : start_time(std::chrono::steady_clock::now()),
                           last_report_time(std::chrono::steady_clock::now()) {}
    };

    MonitoringStats stats_;
    std::unique_ptr<std::thread> monitor_thread_;

    void MonitorLoop();

public:
    StreamHandler();
    ~StreamHandler();

    // Single request-response (inherited from original implementation)
    Status WriteBatch(ServerContext* context,
                     const WriteRequest* request,
                     WriteResponse* response) override;

    // Client-side streaming
    Status WriteStream(ServerContext* context,
                      ServerReader<WriteRequest>* reader,
                      WriteResponse* response) override;

    // Columnar batch processing
    Status WriteColumnar(ServerContext* context,
                        const kwdb::ingest::v1::ColumnarBatch* request,
                        WriteResponse* response) override;

    // Bidirectional streaming with progress reporting
    Status WriteBidirectional(ServerContext* context,
                             ServerReaderWriter<kwdb::ingest::v1::WriteProgress, WriteRequest>* stream) override;

private:
    // Background writer thread
    void WriterLoop();

    // Flush batch to database
    void FlushBatch(std::vector<Point>& batch);

    // Format timestamp
    std::string FormatTimestamp(int64_t ns);

    // SQL escaping functions
    std::string QuoteIdentifier(const std::string& s);
    std::string QuoteLiteral(const std::string& s);

    // Execute SQL
    int ExecuteSQL(const std::string& sql, std::string* err = nullptr);
};